package ar.com.santiagobasulto.crawler.core;

import java.net.URL;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

public class URLProcessorManager implements Runnable{

	public static int NOT_PROCESSED_URL_MAX_CAPACITY = 1000;
	public static int PROCESSED_URL_MAX_CAPACITY = 100000;

	public static int MAX_PROCESSOR_THREADS = 10;

	private BlockingQueue<URL> notProcessedUrls = new ArrayBlockingQueue<URL>(NOT_PROCESSED_URL_MAX_CAPACITY);
	private Set<URL> processedUrls = new HashSet<URL>(PROCESSED_URL_MAX_CAPACITY);

	private ExecutorService urlProcesspool;

	private Class<? extends URLProcessor> processorClass;

	public URLProcessorManager(Class<? extends URLProcessor> processorClass){
		this.processorClass = processorClass;
		urlProcesspool = Executors.newFixedThreadPool(POOL_SIZE);
	}

	@Override
	public void run() {
		for(int i=0;i<BURST_NUMBER.length;i++){
			for(int j=0;j<BURST_NUMBER[i];j++)
				urlProcesspool.submit(new URLProcessorWrapper(notProcessedUrls,processedUrls,processorClass,urlProcesspool));
			try {
				Thread.sleep(TIME_TO_SLEEP_AFTER_BURST[i]);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				//e.printStackTrace();
			}
		}
		boolean done = false;
		while(!done){
			ThreadPoolExecutor pool = (ThreadPoolExecutor)urlProcesspool;
			boolean empty = false;
			while(!empty){
				int count = pool.getActiveCount();
				if(count==0){
					empty=true;
				}else{
					/*if(count<50){
						for (int i = 0; i < LOW_POOL_SIZE; i++)
							urlFinderpool.submit(new URLFinder(notCrawledUrls,crawledUrls));
					}*/
				}
			}
			if(empty){
				done=true;
				WebCrawlerLogger.log.info("No existen mas URLs que procesar");
				urlProcesspool.shutdown();
			}
		}
	}

	public void offer(URL url){
		synchronized (processedUrls) {
			if(!processedUrls.contains(url)){
				notProcessedUrls.add(url);
			}
		}
	}

	private static int POOL_SIZE = 300;

	private static int LOW_POOL_SIZE = 20;

	private static int[] TIME_TO_SLEEP_AFTER_BURST = {1000,1000,1000,1000,1000};
	private static int[] BURST_NUMBER = {10,10,10,10,10};

}
